package org.multi.core;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.StringJoiner;
import java.util.function.BiConsumer;
import java.util.function.BinaryOperator;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.function.ToDoubleFunction;
import java.util.function.ToIntFunction;
import java.util.function.ToLongFunction;
import java.util.stream.Collector;
import org.multi.core.state.DoubleState;
import org.multi.core.state.IntState;
import org.multi.core.state.LongState;

public interface Reducer<T, A, R> {

    @SuppressWarnings("unchecked")
    static <I, R> Function<I, R> castingIdentity() {
        return i -> (R) i;
    }

    static <T, A, R> Reducer<T, A, R> of(Supplier<A> supplier, BiConsumer<A, T> accumulator, Function<A, R> finisher) {
        return new Reducers<>(supplier, accumulator, finisher);
    }

    static <T, A, R> Reducer<T, A, R> of(Supplier<A> supplier, BiConsumer<A, T> accumulator) {
        return new Reducers<>(supplier, accumulator, castingIdentity());
    }

    static <T, A, R> Reducer<T, A, R> of(Collector<T, A, R> collectors) {
        return new Reducers<>(collectors.supplier(), collectors.accumulator(), collectors.combiner(), collectors.finisher());
    }

    static <T, A extends Collection<T>> Reducer<T, A, A> collect(Supplier<A> supplier) {
        return of(supplier, Collection::add);
    }

    static <T> Reducer<T, List<T>, List<T>> toList() {
        return collect(ArrayList::new);
    }

    static <T> Reducer<T, Set<T>, Set<T>> toSet() {
        return collect(LinkedHashSet::new);
    }

    static <T> Reducer<T, List<T>, List<T>> toList(int initialCapacity) {
        return collect(() -> new ArrayList<>(initialCapacity));
    }

    static <T> Reducer<T, Set<T>, Set<T>> toSet(int initialCapacity) {
        return collect(() -> new LinkedHashSet<>(initialCapacity));
    }

    static <T> Reducer<T, IntState, Integer> count() {
        return of(() -> new IntState(0), (state, item) -> state.value++, (state) -> state.value);
    }

    static <T> Reducer<T, IntState, Integer> count(Predicate<T> predicate) {
        return of(() -> new IntState(0), (state, item) -> {
            if (predicate.test(item)) {
                state.value++;
            }
        }, (state) -> state.value);
    }

    static <T> Reducer<T, List<T>, List<T>> sort() {
        return sort(null);
    }

    static <T> Reducer<T, List<T>, List<T>> sort(Comparator<T> comparator) {
        return of(ArrayList::new, List::add, (list) -> {
            list.sort(comparator);
            return list;
        });
    }

    static <T> Reducer<T, List<T>, List<T>> sortDesc(Comparator<T> comparator) {
        return of(ArrayList::new, List::add, (list) -> {
            list.sort(comparator.reversed());
            return list;
        });
    }

    static <T> Reducer<T, StringJoiner, String> join(String delimiter, String prefix, String suffix, Function<T, String> mapper) {
        return of(() -> new StringJoiner(delimiter, prefix, suffix), (state, item) -> state.add(mapper.apply(item)),
                StringJoiner::toString);
    }

    static <T> Reducer<T, StringJoiner, String> join(String delimiter, Function<T, String> mapper) {
        return of(() -> new StringJoiner(delimiter), (state, item) -> state.add(mapper.apply(item)), StringJoiner::toString);
    }

    static <T> Reducer<T, StringJoiner, String> join(String delimiter) {
        return join(delimiter, "", "", T::toString);
    }

    static <T> Reducer<T, DoubleState, Double> sum(ToDoubleFunction<T> mapper) {
        return of(() -> new DoubleState(0), (state, item) -> state.value += mapper.applyAsDouble(item), (state) -> state.value);
    }

    static <T> Reducer<T, IntState, Integer> sumInt(ToIntFunction<T> mapper) {
        return of(() -> new IntState(0), (state, item) -> state.value += mapper.applyAsInt(item), (state) -> state.value);
    }

    static <T> Reducer<T, LongState, Long> sumLong(ToLongFunction<T> mapper) {
        return of(() -> new LongState(0), (state, item) -> state.value += mapper.applyAsLong(item), (state) -> state.value);
    }

    static <T> Reducer<T, LinkedHashSet<T>, Multi<T>> distinct() {
        return of(LinkedHashSet::new, LinkedHashSet::add, (state) -> state::forEach);
    }

    /**
     * 提供初始值
     */
    Supplier<A> supplier();

    /**
     * 让流中元素与初始值进行处理
     */
    BiConsumer<A, T> accumulator();

    /**
     * 合并两个值
     */
    BinaryOperator<A> combiner();

    /**
     * 最后处理返回值
     */
    Function<A, R> finisher();

    class Reducers<T, A, R> implements Reducer<T, A, R> {
        Supplier<A> supplier;
        BiConsumer<A, T> accumulator;
        BinaryOperator<A> combiner;
        Function<A, R> finisher;

        public Reducers(Supplier<A> supplier, BiConsumer<A, T> accumulator, Function<A, R> finisher) {
            this.supplier = supplier;
            this.accumulator = accumulator;
            this.finisher = finisher;
        }

        public Reducers(Supplier<A> supplier, BiConsumer<A, T> accumulator, BinaryOperator<A> combiner, Function<A, R> finisher) {
            this.supplier = supplier;
            this.accumulator = accumulator;
            this.combiner = combiner;
            this.finisher = finisher;
        }

        @Override
        public Supplier<A> supplier() {
            return supplier;
        }

        @Override
        public BiConsumer<A, T> accumulator() {
            return accumulator;
        }

        @Override
        public BinaryOperator<A> combiner() {
            return combiner;
        }

        @Override
        public Function<A, R> finisher() {
            return finisher;
        }
    }
}
